package com.bw.medical.realtime.app.dwd;

import com.bw.medical.realtime.common.MedicalCommon;
import com.bw.medical.realtime.util.CreateEnvUtil;
import com.bw.medical.realtime.util.KafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/*
9.6 用户域患者添加事务事实表
 */

public class DwdUserPatientAdd {
    public static void main(String[] args) {
        // TODO 1. 初始化表处理环境
        StreamExecutionEnvironment env = CreateEnvUtil.getStreamEnv(8087, "medical_dwd_user_patient_add");
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        env.setParallelism(1);

        // TODO 2. 读取数据
        String topic = MedicalCommon.KAFKA_ODS_TOPIC;
        String groupId = "medical_dwd_user_patient_add";
        CreateEnvUtil.createOdsTable(tableEnv, topic, groupId);

        // TODO 3. 筛选数据
        Table resultTable = tableEnv.sqlQuery("select \n" +
                "`data`['id'] `id`,\n" +
                "`data`['create_time'] `add_time`, \n" +
                "`data`['birthday'] `birthday`,\n" +
                "`data`['gender'] `gender_code`,\n" +
                "concat(substr(`data`['name'], 1, 1), '*') `name`,\n" +
                "`data`['user_id'] `user_id`\n" +
                "from `topic_db`\n" +
                "where `table` = 'patient'\n" +
                "and type = 'insert'\n");
        tableEnv.createTemporaryView("result_table", resultTable);

        // TODO 4. 写入 Kafka 目标主题
        String sinkTopic = "medical_dwd_user_patient_add";
        tableEnv.executeSql("create table " + sinkTopic + "(\n" +
                "`id` string,\n" +
                "`add_time` string, \n" +
                "`birthday` string,\n" +
                "`gender_code` string,\n" +
                "`name` string,\n" +
                "`user_id` string\n" +
                ") " + KafkaUtil.getKafkaSinkDDL(sinkTopic));
        tableEnv.executeSql("insert into " + sinkTopic + " select * from result_table");
    }
}
